package com.wuwangfu.func;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Supplier;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-06-29
 */
public class AsyncHttpFunc extends RichAsyncFunction<String, Tuple2<String,String>> {
    //异步请求的httpClient
    private transient CloseableHttpAsyncClient httpclient;
    //请求的url地址
    private String url;
    //异步httpClient支持的最大连接
    private int maxConnTotal;

    public AsyncHttpFunc(String url, int maxConnTotal) {
        this.url = url;
        this.maxConnTotal = maxConnTotal;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        RequestConfig requestConfig = RequestConfig.custom().build();
        httpclient = HttpAsyncClients.custom()
                .setMaxConnTotal(maxConnTotal)//创建HttpAsyncClients请求连接池
                .setDefaultRequestConfig(requestConfig).build();
        //启动异步请求httpclient
        httpclient.start();
    }

    @Override
    public void asyncInvoke(String uid, ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
        //请求的地址和参数
        HttpGet httpGet = new HttpGet(url + "/?uid=" + uid);
        //执行请求返回future
        Future<HttpResponse> future = httpclient.execute(httpGet, null);
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    HttpResponse response = future.get();
                    //调用Future的get方法获取请求的结果
                    String res = null;
                    if (response.getStatusLine().getStatusCode() == 200){
                        res = EntityUtils.toString(response.getEntity());
                    }
                    return res;
                } catch (Exception e) {
                    return null;
                }
            }
        }).thenAccept((String result)->{
            //将结果添加到resultFuture中输出（complete方法的参数只能为集合，如果只有一个元素，就返回一个单例集合）
            resultFuture.complete(Collections.singleton(Tuple2.of(uid,result)));
        });
    }

    @Override
    public void close() throws Exception {
        //关闭HttpAsyncClients请求连接池
        httpclient.close();
    }
}
